#include <basic/errcode.h>
//#include <basic/shm.h>
#include <basic/mqe.h>

//#define _SWITCH_SHM_LIMIT   KB(8)   // 消息转由共享内存传递的限额
//#define _SHM_UNIT_NUM       10      // 共享内存中消息条数

struct _msg_unit
{/*{{{*/
    long mtype;
    struct _mtext
    {
        char self[FILE_NM_SIZE+1];
        unsigned int syn;
        int buf_size;
        unsigned char buf[MQE_MSG_MAX_SIZE];
//        unsigned int in_shm;
//        int shm_index;
    } mtext;
};/*}}}*/

//struct _shm_unit
//{/*{{{*/
//    enum { _FREE, _USED } state;
//    unsigned char buf[MQE_MSG_MAX_SIZE];
//};/*}}}*/

struct _mqe
{/*{{{*/
    int fd;
    char self[FILE_NM_SIZE+1];
    long self_hash;
    unsigned int syn;
    struct _msg_unit *msg_unit;
    int msg_unit_head_size;
//    struct _shm_unit *shm_unit;
};/*}}}*/

static int _qflag=
IPC_CREAT | S_IRUSR | S_IRGRP | S_IROTH | S_IWUSR | S_IWGRP | S_IWOTH;

//int mqe_create(char *queue)
//{/*{{{*/
//    char area_file[FILE_NM_SIZE+1];
//    struct _shm_unit *shm_unit;
//    int i;
//
//    if(!queue)
//        return(RET_ERR_PARA);
//
//    if(mqe_clear(queue))
//        return(-1);
//
//    sprintf(area_file, "mqe_%s", queue);
//    if(shm_area_create(area_file,
//                sizeof(struct _shm_unit)*_SHM_UNIT_NUM+32,
//                SHM_AREA_LOCK))
//        return(-2);
//    if((shm_unit=(struct _shm_unit *)shm_area_attach(area_file)) ==
//            NULL)
//        return(-3);
//    for(i=0; i<_SHM_UNIT_NUM; i++)
//        shm_unit[i].state=_FREE; 
//    shm_area_detach(shm_unit);
//
//    return(0);
//}/*}}}*/

int mqe_clear(char *queue)
{/*{{{*/
    key_t key;
    int fd;
    int size;
    char buf[BUFFER_SIZE];

    if(!queue)
        return(RET_ERR_PARA);

    key=(key_t)get_hash((unsigned char *)queue, strlen(queue));
    if((fd=msgget(key, _qflag)) == -1)
        return(-1);
    while(1)
    {
        if((size=msgrcv(fd, buf, sizeof(buf), 0,
                        MSG_NOERROR | IPC_NOWAIT)) == -1)
        {
            if(get_last_error(NULL) == ENOMSG)
                break;
            return(-2);
        }
    }

    return(0);
}/*}}}*/

int mqe_attach(char *queue, char *self, MQE *pmqe)
{/*{{{*/
    MQE mqe;
    key_t key;

    if(!pmqe)
        return(RET_ERR_PARA);
    *pmqe=NULL;
    if(!queue || !self || strlen(self) > FILE_NM_SIZE)
        return(RET_ERR_PARA);

    ALLOC_RETERR(mqe, struct _mqe, RET_ERR_ALLOC);
    mqe->msg_unit=NULL;
    key=(key_t)get_hash((unsigned char *)queue, strlen(queue));
    if((mqe->fd=msgget(key, _qflag)) == -1)
    {
        free(mqe);
        return(-1);
    }
    strcpy(mqe->self, self);
    mqe->self_hash=labs((long)get_hash((unsigned char *)self, strlen(self)));
    mqe->syn=1;
    ALLOC_DOERR(mqe->msg_unit, struct _msg_unit,
            free(mqe); return(RET_ERR_ALLOC));
    mqe->msg_unit_head_size=(int)(
            (char *)mqe->msg_unit->mtext.buf -
            (char *)mqe->msg_unit);
//    if(flag & MQE_SHM)
//    {
//        char area_file[FILE_NM_SIZE+1];
//        sprintf(area_file, "mqe_%s", queue);
//        if((mqe->shm_unit=(struct _shm_unit *)shm_area_attach(
//                        area_file)) == NULL)
//        {
//            free(mqe->msg_unit);
//            free(mqe);
//            return(-2);
//        }
//    }
//    else
//    {
//        mqe->shm_unit=(struct _shm_unit *)NULL;
//    }

    *pmqe=mqe;
    return(0);
}/*}}}*/

int mqe_detach(MQE mqe)
{/*{{{*/
    if(!mqe)
        return(RET_ERR_PARA);

    free(mqe->msg_unit);
//    shm_area_detach((void *)mqe->shm_unit);
    free(mqe);
    return(0);
}/*}}}*/

static int _send(MQE mqe, char *target, unsigned int syn,
        unsigned char *msg, int msg_size)
{/*{{{*/
    struct _mtext *mtext;

    mtext=&mqe->msg_unit->mtext;
    mqe->msg_unit->mtype=labs((long)get_hash((unsigned char *)target,
                strlen(target)));
    strcpy(mtext->self, mqe->self);
    mtext->syn=syn;
    mtext->buf_size=msg_size;
    memcpy(mtext->buf, msg, msg_size);
//    if(mqe->shm_unit)
//    {
//        if(msg_size > _SWITCH_SHM_LIMIT)
//        {
//            struct _shm_unit *shm_unit=mqe->shm_unit;
//            int shm_index;
//
//            shm_area_enter((void *)shm_unit);
//            for(shm_index=0; shm_index<_SHM_UNIT_NUM; shm_index++)
//                if(shm_unit[shm_index].state == _FREE)
//                    break;
//            if(shm_index == _SHM_UNIT_NUM)
//            {
//                mtext->in_shm=0;
//            }
//            else
//            {
//                mtext->in_shm=1;
//                shm_unit[shm_index].state=_USED;
//                memcpy(shm_unit[shm_index].buf, msg, msg_size);
//            }
//            shm_area_leave((void *)shm_unit);
//        }
//        else
//            mtext->in_shm=0;
//    }
//    else
//        mtext->in_shm=0;
//    if(mtext->in_shm)
//        msg_size=0;
//    else
//        memcpy(mtext->buf, msg, msg_size);
    if(msgsnd(mqe->fd, (void *)mqe->msg_unit,
                mqe->msg_unit_head_size+msg_size, 0) == -1)
    {
        if(get_last_error(NULL) == EINTR)
            return(RET_ERR_TIMEOUT);
        return(-1);
    }

    return(0);
}/*}}}*/

int mqe_send(MQE mqe, char *target, unsigned char *msg, int msg_size)
{/*{{{*/
    if(!mqe || !target ||
            !msg || msg_size < 0 || msg_size > MQE_MSG_MAX_SIZE)
        return(RET_ERR_PARA);
    return(_send(mqe, target, 0, msg, msg_size));
}/*}}}*/

static int _recv(MQE mqe, char *source, unsigned int *syn,
        unsigned char *msg, int *msg_size)
{/*{{{*/
    struct _mtext *mtext;
    int mtext_size;
    int buf_size;

    if((mtext_size=msgrcv(mqe->fd, (void *)mqe->msg_unit,
           MQE_MSG_MAX_SIZE, mqe->self_hash, 0)) == -1)
    {
        if(get_last_error(NULL) == EINTR)
            return(RET_ERR_TIMEOUT);
        return(-1);
    }
    mtext=&mqe->msg_unit->mtext;
    mtext_size-=mqe->msg_unit_head_size;
    if(mtext_size < 0)
        return(-2);
    if((buf_size=mtext->buf_size) < 0)
        return(-3);
    if(buf_size != mtext_size)
        return(-4);
    memcpy(msg, mtext->buf, buf_size);
//    if(mtext->in_shm)
//    {
//        struct _shm_unit *shm_unit=mqe->shm_unit;
//        int shm_index=mtext->shm_index;
//
//        if(shm_index < 0 || shm_index > _SHM_UNIT_NUM)
//            return(-4);
//        shm_area_enter((void *)shm_unit);
//        memcpy(msg, shm_unit[shm_index].buf, buf_size);
//        shm_unit[shm_index].state=_FREE;
//        shm_area_leave((void *)shm_unit);
//    }
//    else
//    {
//        if(buf_size != mtext_size)
//            return(-5);
//        memcpy(msg, mtext->buf, buf_size);
//    }
    if(source)  strcpy(source, mtext->self);
    if(syn)  *syn=mtext->syn;
    msg[buf_size]=ZERO;
    *msg_size=buf_size;

    return(0);
}/*}}}*/

int mqe_recv(MQE mqe, char *source, unsigned char *msg, int *msg_size)
{/*{{{*/
    if(!mqe || !msg || !msg_size)
        return(RET_ERR_PARA);
    return(_recv(mqe, source, NULL, msg, msg_size));
}/*}}}*/

int mqe_syn_call(MQE mqe, char *target,
    unsigned char *request, int request_size,
    unsigned char *respond, int *respond_size)
{/*{{{*/
    int result;
    unsigned int syn;

    if(!mqe || !target ||
            !request || request_size < 0 ||
            request_size > MQE_MSG_MAX_SIZE ||
            !respond || !respond_size)
        return(RET_ERR_PARA);

    if(++mqe->syn == 0)
        mqe->syn=1;
    if((result=_send(mqe, target, mqe->syn, request,
                    request_size)))
        return(-100+result);
    do
    {
        if((result=_recv(mqe, NULL, &syn, respond, respond_size)))
            return(-200+result);
    } while(mqe->syn != syn);

    return(0);
}/*}}}*/

int mqe_syn_fetch(MQE mqe, MQE_SYN_ATTR *attr,
        unsigned char *msg, int *msg_size)
{/*{{{*/
    if(!attr)
        return(RET_ERR_PARA);
    return(_recv(mqe, attr->source, &attr->syn, msg, msg_size));
}/*}}}*/

int mqe_syn_reply(MQE mqe, MQE_SYN_ATTR *attr,
        unsigned char *msg, int msg_size)
{/*{{{*/
    if(!attr)
        return(RET_ERR_PARA);
    return(_send(mqe, attr->source, attr->syn, msg, msg_size));
}/*}}}*/
